package zookeeper.demo2;


import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.retry.RetryForever;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Slf4j
public class ZkDiscoveryService {
    private String zkUrl = "localhost:2181";
    private Integer zkRetryInterval = 3;
    private Integer zkConnectionTimeout = 3000;
    private Integer zkSessionTimeout = 3000;
    private String zkDir = "/test/lin/rule/index";

    private CuratorFramework client;

    private NodeCache nodeCache;

    @PostConstruct
    public void init() {

        log.info("Initializing discovery service using ZK connect string: {}", zkUrl);
        try {
            client = CuratorFrameworkFactory.newClient(zkUrl, zkSessionTimeout, zkConnectionTimeout, new RetryForever(zkRetryInterval));
            client.start();
            client.blockUntilConnected();
            nodeCache = new NodeCache(client, zkDir, false);
            initIndexNode();
            nodeCache.start();

            nodeCache.getListenable().addListener(() -> {
                byte[] nodeData = nodeCache.getCurrentData().getData();
                //当zk监听到更新时，记录下最新的日志id
                log.info("节点发生变化 newData {}", new String(nodeData));
                System.out.println(new String(nodeData));
            });

        } catch (Exception e) {
            log.error("Failed to connect to ZK: {}", e.getMessage(), e);
            CloseableUtils.closeQuietly(client);
            throw new RuntimeException(e);
        }
    }

    public void updateValue(String value) throws Exception {
        try {
            client.setData().forPath(zkDir, value.getBytes());
        } catch (KeeperException.NoNodeException e) {
            initIndexNode();
        }
    }

    @PreDestroy
    public void destroy() {
        CloseableUtils.closeQuietly(client);
        log.info("Stopped discovery service");

    }

    private void initIndexNode() throws Exception {
        if (client.checkExists()
                .forPath(zkDir) == null) {
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT)
                    .forPath(zkDir);
        }
    }

}
